package cn.ucox.web.ms.persistence;

import cn.ucox.web.ms.Bootstarpable;
import cn.ucox.web.ms.Constant;
import cn.ucox.web.ms.redis.RedisClient;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Base64Utils;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static cn.ucox.web.ms.Constant.CPU_PROCESSOR_NUMBER;
import static cn.ucox.web.ms.Constant.SYNC_TRANS_RESULT_DATA_BACKUP;

/**
 * 持久化数据队列监听，解析程序将从FTP中获取的数据解析后，
 * <br>推送至持久化数据队列中。持久化组件读取队列数据生成DDL语句
 *
 * @author chenw
 * @create 2017-06-06 10:39
 * @email javacspring@gmail.com
 */
public class PersistenceQueueListener implements Bootstarpable {

    private static final Logger logger = LoggerFactory.getLogger(PersistenceQueueListener.class);

    private ExecutorService service = Executors.newFixedThreadPool(CPU_PROCESSOR_NUMBER * 2);
    private QueueListener listener;
    private static RedisClient redisClient;


    private IDataPersistence dataPersistence;

    private IAttachmentPersistence attachmentPersistence;

    static {
        redisClient = RedisClient.instance();
    }

    @Override
    public void start() {
        listener = new QueueListener();
        new Thread(listener).start();
    }

    @Override
    public void stop() {
        listener.stop();
    }

    private class QueueListener implements Runnable {

        volatile boolean runningFlag = true;

        QueueListener() {
            this.runningFlag = true;
        }

        void stop() {
            runningFlag = false;
            logger.info("数据队列监听停止");
        }

        @Override
        public void run() {
            logger.debug("启动持久化数据监听列队");
            while (runningFlag) {
                SyncRowData rowData = null;
                String data = null;
                try {
                    data = redisClient.rightPop(Constant.SYNC_TRANS_RESULT_DATA);
                    if (null != data && 0 != data.length()) {
                        logger.debug("取出队列数据{}", data.substring(0, 50));
                        //1.将数据反序列化为JSON
                        rowData = JSON.parseObject(data, SyncRowData.class);
                        // base64转String
                        byte[] dataStr = Base64Utils.decodeFromString(rowData.getData());
                        rowData.setData(new String(dataStr));
                        if (rowData.isAtt()) {//附件类型记录
                            attachmentPersistence.persistence(rowData);
                        } else {
                            dataPersistence.persistence(rowData);
                        }
                    }else {
                        logger.debug("持久化队列数据为空");
                        delayTime();
                    }
                } catch (Exception ex) {
                    //TODO: 1.拿到同步ID、表名、时间、同步失败原因(当前是持久化异常)、异常信息日志
                    //TODO: 2.记录到同步失败日志队列(SYNC_LOG)
                    //TODO: 3.定时任务将失败日志队列数据持久化数据库同步失败日志表

                    JSONObject object = new JSONObject();
                    if (null != rowData) {
                        //失败记录备份至redis,其中业务data数据不存
                        rowData.setData(null);
                        object = JSON.parseObject(JSONObject.toJSONString(rowData));
                    }else {
                        //转JSON失败，将data存入异常数据备份队列
                        object.put("bizData", data.substring(0, 100));
                    }
                    //将错误信息记录至redis中
                    object.put("errorMsg", ex.getMessage());

                    //将错误类名 行数 方法名记录至redis中
                    String stackTrace = "";
                    for (StackTraceElement stackTraceElement : ex.getStackTrace()) {
                        if (stackTraceElement.getClassName().contains("cn.ucox.web.ms")) {
                            stackTrace = stackTrace + stackTraceElement.getClassName() + "&" + stackTraceElement.getMethodName() + " " + stackTraceElement.getLineNumber() + ";";
                        }
                    }
                    object.put("stackTrace", stackTrace);

                    redisClient.leftPush(SYNC_TRANS_RESULT_DATA_BACKUP, object.toJSONString());
                    logger.error("持久化数据异常,同步ID:{}，异常信息:{}", null != rowData ? rowData.getId() : "null,解析错误", ex.getMessage());
                }
            }

        }
    }

    private void delayTime() {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException ex) {
            logger.error("延时错误", ex);
        }
    }

    public void setDataPersistence(IDataPersistence dataPersistence) {
        this.dataPersistence = dataPersistence;
    }

    public void setAttachmentPersistence(IAttachmentPersistence attachmentPersistence) {
        this.attachmentPersistence = attachmentPersistence;
    }
}